/*
 * Licensed to Diennea S.r.l. under one
 * or more contributor license agreements. See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership. Diennea S.r.l. licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 *
 */
package herddb.core;

import herddb.index.IndexOperation;
import herddb.index.KeyToPageIndex;
import herddb.log.CommitLog;
import herddb.log.LogSequenceNumber;
import herddb.model.Index;
import herddb.model.StatementEvaluationContext;
import herddb.model.StatementExecutionException;
import herddb.model.TableContext;
import herddb.model.Transaction;
import herddb.storage.DataStorageManager;
import herddb.storage.DataStorageManagerException;
import herddb.utils.Bytes;
import herddb.utils.ILocalLockManager;
import herddb.utils.LocalLockManager;
import java.util.AbstractMap.SimpleImmutableEntry;
import java.util.List;
import java.util.Map;
import java.util.stream.Stream;

/**
 * Index runtime
 *
 * @author enrico.olivelli
 */
public abstract class AbstractIndexManager implements AutoCloseable {

    protected final Index index;
    protected final AbstractTableManager tableManager;
    protected final DataStorageManager dataStorageManager;
    protected final String tableSpaceUUID;
    protected final CommitLog log;
    /**
     * This value is not empty until the transaction who creates the table does not commit
     */
    protected long createdInTransaction;
    private final boolean unique;
    private final ILocalLockManager lockManager;

    public AbstractIndexManager(Index index, AbstractTableManager tableManager,
                                             DataStorageManager dataStorageManager, String tableSpaceUUID,
                                             CommitLog log, long createdInTransaction,
                                             int writeLockTimeout, int readLockTimeout) {
        this.index = index;
        this.createdInTransaction = createdInTransaction;
        this.tableManager = tableManager;
        this.dataStorageManager = dataStorageManager;
        this.tableSpaceUUID = tableSpaceUUID;
        this.log = log;
        this.unique = index.unique;
        if (unique) {
            LocalLockManager newLockManager = new LocalLockManager();
            newLockManager.setWriteLockTimeout(writeLockTimeout);
            newLockManager.setReadLockTimeout(readLockTimeout);
            this.lockManager = newLockManager;
        } else {
            lockManager = null;
        }
    }

    public final boolean isUnique() {
        return unique;
    }

    public ILocalLockManager getLockManager() {
        return lockManager;
    }

    public final Index getIndex() {
        return index;
    }

    public final String getIndexName() {
        return index.name;
    }

    public final String[] getColumnNames() {
        return index.columnNames;
    }

    /**
     * Boots the index, this method usually reload state from the DataStorageManager
     *
     * @param sequenceNumber sequence number from which boot the index
     * @throws DataStorageManagerException
     */
    public void start(LogSequenceNumber sequenceNumber) throws DataStorageManagerException {

        boolean needRebuild = !doStart(sequenceNumber);

        if (needRebuild) {
            dropIndexData();
            rebuild();
        }
    }

    /**
     * Boots the index, this method usually reload state from the DataStorageManager
     *
     * @param sequenceNumber sequence number from which boot the index
     * @return {@code false} if index cannot be started and need a rebuild, @code true} otherwise
     * @throws DataStorageManagerException
     */
    protected abstract boolean doStart(LogSequenceNumber sequenceNumber) throws DataStorageManagerException;

    /**
     * Release resources. Do not drop data
     */
    @Override
    public void close() {

    }

    /**
     * Rebuild entirely the index, usually performing a full table scan
     *
     * @throws DataStorageManagerException
     */
    public abstract void rebuild() throws DataStorageManagerException;

    /**
     * Ensures that all data is persisted from disk
     *
     * @param sequenceNumber
     * @return
     * @throws DataStorageManagerException
     */
    public abstract List<PostCheckpointAction> checkpoint(LogSequenceNumber sequenceNumber, boolean pin) throws DataStorageManagerException;

    /**
     * Unpin a previously pinned checkpont (see {@link #checkpoint(LogSequenceNumber, boolean)})
     *
     * @throws DataStorageManagerException
     */
    public abstract void unpinCheckpoint(LogSequenceNumber sequenceNumber) throws DataStorageManagerException;

    /**
     * Basic function of the index. The index returns the list of PKs of the table which match the predicate (IndexOperation) Beare that this function could return a super set of the list of the PKs
     * which actually match the predicate. TableManager will check every record againts the (WHERE) Predicate in order to ensure the final result
     *
     * @param operation
     * @param context
     * @param tableContext
     * @return a stream on the PK values of the tables which match the index
     * @throws StatementExecutionException
     */
    protected abstract Stream<Bytes> scanner(IndexOperation operation, StatementEvaluationContext context, TableContext tableContext) throws StatementExecutionException;

    /**
     * This function is called from the TableManager to perform scans. It usually have to deal with a JOIN on the KeyToPageIndex of the TableManager
     *
     * @param operation
     * @param context
     * @param tableContext
     * @param keyToPageIndex
     * @return
     * @throws DataStorageManagerException
     * @throws StatementExecutionException
     */
    public Stream<Map.Entry<Bytes, Long>> recordSetScanner(IndexOperation operation, StatementEvaluationContext context, TableContext tableContext, KeyToPageIndex keyToPageIndex) throws DataStorageManagerException, StatementExecutionException {
        return scanner(operation, context, tableContext).map((b) -> {
            Long idPage = keyToPageIndex.get(b);
            if (idPage == null) {
                return null;
            }
            return ((Map.Entry<Bytes, Long>) new SimpleImmutableEntry<>(b, idPage));
        }).filter(p -> p != null);
    }

    public abstract void recordUpdated(Bytes key, Bytes indexKeyRemoved, Bytes indexKeyAdded) throws DataStorageManagerException;

    public abstract void recordInserted(Bytes key, Bytes indexKey) throws DataStorageManagerException;

    public abstract void recordDeleted(Bytes key, Bytes indexKey) throws DataStorageManagerException;

    /**
     * Drop the index from persist storage
     *
     * @throws DataStorageManagerException
     */
    public void dropIndexData() throws DataStorageManagerException {
        dataStorageManager.dropIndex(tableSpaceUUID, index.uuid);
    }

    /**
     * Truncate the index from persist storage
     * <p>
     * Differs from {@link #dropIndexData()} because it leaves in place every support structure needed by index runtime (for example if index is persisted into a directory it doesn't delete the
     * directory itself). The index will continue to work but without current data that will be erased
     * </p>
     *
     * @throws DataStorageManagerException
     */
    public void truncateIndexData() throws DataStorageManagerException {
        dataStorageManager.truncateIndex(tableSpaceUUID, index.uuid);
    }

    final void onTransactionCommit(Transaction transaction, boolean recovery) throws DataStorageManagerException {
        if (createdInTransaction > 0) {
            if (transaction.transactionId != createdInTransaction) {
                throw new DataStorageManagerException("this indexManager (" + getIndexName() + ") is available only on transaction " + createdInTransaction
                        + " and not in transaction " + transaction.transactionId);
            }
            createdInTransaction = 0;
        }
        if (lockManager != null) {
            transaction.releaseLocksOnTable(getIndexName(), lockManager);
        }
    }


    final void onTransactionRollback(Transaction transaction) {
        if (lockManager != null) {
            transaction.releaseLocksOnTable(getIndexName(), lockManager);
        }
    }

    public long getCreatedInTransaction() {
        return createdInTransaction;
    }

    public final boolean isAvailable() {
        return createdInTransaction == 0;
    }

    /**
     * Erase the index. Out-side the scope of a transaction
     */
    public abstract void truncate() throws DataStorageManagerException;

    public abstract boolean valueAlreadyMapped(Bytes key, Bytes primaryKey)  throws DataStorageManagerException;
}
